package com.warren.financial.lease.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.warren.financial.lease.realtime.bean.*;
import com.warren.financial.lease.realtime.common.FinancialLeaseCommon;
import com.warren.financial.lease.realtime.util.CreateEnvUtil;
import com.warren.financial.lease.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @AUTHOR warren
 * @date 2023/12/13
 **/
public class DwdBaseApp {

    public static void main(String[] args) throws Exception {

        //1.初始化流数据
        String appName = "dwd_base_app";
        StreamExecutionEnvironment env = CreateEnvUtil.getExecutionEnvironment(8082, appName);
        env.setParallelism(1);

        //2.从ods曾读取topic_db
        KafkaSource<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(FinancialLeaseCommon.KAFKA_ODS_TOPIC, appName, OffsetsInitializer.earliest());
        DataStreamSource<String> odsStream = env.fromSource(kafkaConsumer, WatermarkStrategy.noWatermarks(), appName);

        //3.筛选出表示业务过程的5张表
        SingleOutputStreamOperator<JSONObject> flatMapStream = odsStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> collector) throws Exception {

                JSONObject jsonObject = JSONObject.parseObject(value);
                String tableName = jsonObject.getString("table");

                //添加etl 八宗衡下游数据是完整
                if ("credit".equals(tableName) || "credit_facility_status".equals(tableName) || "reply".equals(tableName)) {
                    if (jsonObject.getJSONObject("data").getString("credit_facility_id") != null) {
                        collector.collect(jsonObject);
                    }
                } else if ("credit_facility".equals(tableName) && jsonObject.getJSONObject("data").getString("id") != null) {
                    collector.collect(jsonObject);
                } else if ("contract".equals(tableName)) {
                    if (jsonObject.getJSONObject("data").getString("credit_id") != null) {
                        collector.collect(jsonObject);
                    }
                }
            }
        });

        //3.1 筛选出contract 流
        SingleOutputStreamOperator<JSONObject> contractStream = flatMapStream.filter(v -> "contract".equals(v.getString("table")));
//        SingleOutputStreamOperator<JSONObject> contractStream = flatMapStream.filter(new FilterFunction<JSONObject>() {
//            @Override
//            public boolean filter(JSONObject value) throws Exception {
//
//                return "contract".equals(value.getString("table"));
//            }
//        });

        //3.2 筛选出除contract 表
        SingleOutputStreamOperator<JSONObject> mainStream = flatMapStream.filter(v -> !"contract".equals(v.getString("table")));
//        SingleOutputStreamOperator<JSONObject> mainStream = flatMapStream.filter(new FilterFunction<JSONObject>() {
//            @Override
//            public boolean filter(JSONObject value) throws Exception {
//
//                return !"contract".equals(value.getString("table"));
//            }
//        });


        //4 按照授信id进行分组
        KeyedStream<JSONObject, String> keyedStream = mainStream.keyBy(jsonObject -> {
            String tableName = jsonObject.getString("table");
            if ("credit_facility".equals(tableName)) {
                return jsonObject.getJSONObject("data").getString("id");
            }
            return jsonObject.getJSONObject("data").getString("credit_facility_id");
        });

        //keyedStream.print("keyed>===");

        // 5动态分流
        // 5.1 审批拒绝侧输出流标签
        OutputTag<DwdAuditRejectBean> rejectStreamTag = new OutputTag<DwdAuditRejectBean>("reject_stream"){};

        // 5.2 审批取消侧输出流标签
        OutputTag<DwdAuditCancelBean> cancelStreamTag = new OutputTag<DwdAuditCancelBean>("cancel_stream"){};

        // 5.3 授信域 测输出流标签
        OutputTag<DwdCreditAddBean> creditAddBeanTag = new OutputTag<DwdCreditAddBean>("credit_add_stream"){};

        OutputTag<DwdCreditOccupyBean> occupyBeanTag = new OutputTag<DwdCreditOccupyBean>("credit_occupy_stream") {};

        // 5.4 租赁域 侧输出流标签
        OutputTag<DwdLeaseContractProducedBean> contractProducedBeanTag = new OutputTag<DwdLeaseContractProducedBean>("contract_produced_stream"){};

        SingleOutputStreamOperator<DwdAuditApprovalBean> processStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, DwdAuditApprovalBean>() {

            // 审批信审经办员工ID
            private ValueState<String> auditManIdStatus;
            // 审批回复
            private ValueState<DwdAuditReplyBean> replyBeanValueState;
            // 审批通过信息记录
            private ValueState<DwdAuditApprovalBean> approvalBeanValueState;
            // 新增授信
            private ValueState<DwdCreditAddBean> creditAddBeanValueState;
            // 完成授信
            private ValueState<DwdCreditOccupyBean> creditOccupyBeanValueState;
            // 合同制作
            private ValueState<DwdLeaseContractProducedBean> leaseContractProducedBeanValueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                auditManIdStatus = getRuntimeContext().getState(new ValueStateDescriptor<String>("audit_man_id", String.class));
                replyBeanValueState = getRuntimeContext().getState(new ValueStateDescriptor<DwdAuditReplyBean>("audit_reply_status", DwdAuditReplyBean.class));
                approvalBeanValueState = getRuntimeContext().getState(new ValueStateDescriptor<DwdAuditApprovalBean>("audit_approval_status", DwdAuditApprovalBean.class));

                // 设置存活时间
                creditAddBeanValueState = getRuntimeContext().getState(new ValueStateDescriptor<DwdCreditAddBean>("credit_add_status", DwdCreditAddBean.class));
                ValueStateDescriptor<DwdCreditAddBean> audit_occupy_status = new ValueStateDescriptor<>("audit_occupy_status", DwdCreditAddBean.class);
                audit_occupy_status.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(30L)).build());
                creditOccupyBeanValueState = getRuntimeContext().getState(new ValueStateDescriptor<DwdCreditOccupyBean>("credit_occupy_status", DwdCreditOccupyBean.class));

                leaseContractProducedBeanValueState = getRuntimeContext().getState(new ValueStateDescriptor<DwdLeaseContractProducedBean>("contract_producer_status", DwdLeaseContractProducedBean.class));
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, DwdAuditApprovalBean>.Context ctx, Collector<DwdAuditApprovalBean> out) throws Exception {
                String tableName = jsonObject.getString("table");
                //不同业务表需要进行不同的处理方式
                if ("credit_facility_status".equals(tableName)) {
                    //审批申请状态表
                    //记录信审批经办审批状态
                    //5 审批通过
                    //6 审批不通过 但是申请复议
                    //20 审批不通过
                    String status = jsonObject.getJSONObject("data").getString("status");
                    if ("5".equals(status) || "6".equals(status) || "20".equals(status)) {
                        String employeeId = jsonObject.getJSONObject("data").getString("employee_id");
                        auditManIdStatus.update(employeeId);
                    }
                } else if ("credit_facility".equals(tableName)) {
                    // 判断审批最终的状态
                    // 16 通过
                    // 20 拒绝
                    // 21 取消
                    // System.out.println("json====>"+jsonObject);
                    String type = jsonObject.getString("type");
                    String status = jsonObject.getJSONObject("data").getString("status");
                    if ("update".equals(type)) {
                        switch (status) {
                            case "16": {
                                DwdAuditApprovalBean auditApprovalBean = jsonObject.getObject("data", DwdAuditApprovalBean.class);
                                //补全信息
                                String auditManId = auditManIdStatus.value();
                                auditApprovalBean.setAuditManId(auditManId);
                                auditApprovalBean.setApproveTime(auditManId);
                                auditManIdStatus.clear();

                                //TODO 补全批复信息
                                auditApprovalBean.setApproveTime(jsonObject.getJSONObject("data").getString("update_time"));
                                auditApprovalBean.setApplyAmount(jsonObject.getJSONObject("data").getBigDecimal("credit_amount"));
                                DwdAuditReplyBean replyBean = replyBeanValueState.value();

                                //数据先后问题
                                if (replyBean == null) {
                                    approvalBeanValueState.update(auditApprovalBean);
                                } else {
                                    auditApprovalBean.setReplyId(replyBean.getId());
                                    auditApprovalBean.setReplyAmount(replyBean.getReplyAmount());
                                    auditApprovalBean.setReplyTime(replyBean.getReplyTime());
                                    auditApprovalBean.setIrr(replyBean.getIrr());
                                    auditApprovalBean.setPeriod(replyBean.getPeriod());
                                    // 回复状态去掉
                                    replyBeanValueState.clear();
                                    // 下游也可能需要用到审批通过得数据
                                    approvalBeanValueState.update(auditApprovalBean);
                                    out.collect(auditApprovalBean);

                                    // 如果新增授信得数据先到，已经存在了状态中，这里需要进行处理，将对应得数据写出道测输出流
                                    DwdCreditAddBean creditAddBean = creditAddBeanValueState.value();
                                    if (creditAddBean != null) {
                                        creditAddBean.setApplyAmount(auditApprovalBean.getApplyAmount());
                                        creditAddBean.setReplyAmount(auditApprovalBean.getReplyAmount());
                                        ctx.output(creditAddBeanTag, creditAddBean);
                                        creditAddBeanValueState.clear();
                                    }

                                    // 如果完成授信占用得数据先到 已经存在状态中 这里需要进行处理 将对应得数据写道侧输出流
                                    DwdCreditOccupyBean creditOccupyBean = creditOccupyBeanValueState.value();
                                    if (creditOccupyBean != null) {
                                        creditOccupyBean.setApplyAmount(auditApprovalBean.getApplyAmount());
                                        creditOccupyBean.setReplyAmount(auditApprovalBean.getReplyAmount());
                                        ctx.output(occupyBeanTag, creditOccupyBean);
                                        creditOccupyBeanValueState.clear();
                                    }

                                    // 如果完成合同制作得数据先到 已经存在状态中 这里需要进行处理 将对应得数据写道侧输出流
                                    DwdLeaseContractProducedBean contractProducedBean = leaseContractProducedBeanValueState.value();
                                    if (contractProducedBean != null) {
                                        contractProducedBean.setApplyAmount(auditApprovalBean.getApplyAmount());
                                        contractProducedBean.setReplyAmount(auditApprovalBean.getReplyAmount());
                                        ctx.output(contractProducedBeanTag, contractProducedBean);
                                        leaseContractProducedBeanValueState.clear();
                                    }
                                }

//                                // 下游也可能需要用到审批通过得数据
//                                approvalBeanValueState.update(auditApprovalBean);
//
//
//                                out.collect(auditApprovalBean);
                                break;
                            }
                            case "20": {
                                DwdAuditRejectBean auditRejectBean = jsonObject.getObject("data", DwdAuditRejectBean.class);
                                // TODO 批复信息
                                String auditManId = auditManIdStatus.value();
                                auditRejectBean.setAuditManId(auditManId);
                                auditManIdStatus.clear();
                                auditRejectBean.setRejectTime(jsonObject.getJSONObject("data").getString("update_time"));
                                auditRejectBean.setApplyAmount(jsonObject.getJSONObject("data").getBigDecimal("credit_amount"));
                                ctx.output(rejectStreamTag, auditRejectBean);

                                // 审批拒绝 可能出现复议 需要将之前得状态清空
                                approvalBeanValueState.clear();

                                break;
                            }
                            case "21": {
                                DwdAuditCancelBean auditCancelBean = jsonObject.getObject("data", DwdAuditCancelBean.class);
                                // TODO 批复信息
                                String auditManId = auditManIdStatus.value();
                                auditCancelBean.setAuditManId(auditManId);
                                auditManIdStatus.clear();
                                auditCancelBean.setCancelTime(jsonObject.getJSONObject("data").getString("update_time"));
                                auditCancelBean.setApplyAmount(jsonObject.getJSONObject("data").getBigDecimal("credit_amount"));
                                ctx.output(cancelStreamTag, auditCancelBean);

                                // 审批取消 不再需要后续 状态清空
                                approvalBeanValueState.clear();

                                break;
                            }
                        }
                    }

                } else if ("reply".equals(tableName)) {
                    // 记录批复信息
                    DwdAuditReplyBean auditReplyBean = jsonObject.getObject("data", DwdAuditReplyBean.class);
                    // 手动添加特殊字段名称得数据
                    auditReplyBean.setReplyTime(jsonObject.getJSONObject("data").getString("create_time"));
                    auditReplyBean.setReplyAmount(jsonObject.getJSONObject("data").getBigDecimal("credit_amount"));

                    // 如果审批通过得数据先到 需要在批复数据得位置完成数据得合并及输出
                    DwdAuditApprovalBean auditApprovalBean = approvalBeanValueState.value();
                    if (auditApprovalBean != null) {
                        auditApprovalBean.setReplyId(auditReplyBean.getId());
                        auditApprovalBean.setReplyAmount(auditReplyBean.getReplyAmount());
                        auditApprovalBean.setReplyTime(auditReplyBean.getReplyTime());
                        auditApprovalBean.setIrr(auditApprovalBean.getIrr());
                        auditApprovalBean.setPeriod(auditReplyBean.getPeriod());

                        // 将审批通过得数据当做主流数据写出
                        out.collect(auditApprovalBean);
                        approvalBeanValueState.update(auditApprovalBean);

                        // 此时审批通过得数据菜完整
                        // 如果新增授信得数据先到，已经存在了状态中，这里需要进行处理，将对应得数据写出道测输出流
                        DwdCreditAddBean creditAddBean = creditAddBeanValueState.value();
                        //System.out.println(creditAddBean);
                        if (creditAddBean != null) {
                            creditAddBean.setApplyAmount(auditApprovalBean.getApplyAmount());
                            creditAddBean.setReplyAmount(auditApprovalBean.getReplyAmount());
                            creditAddBean.setAddTime(auditApprovalBean.getApproveTime());
                            ctx.output(creditAddBeanTag, creditAddBean);
                            creditAddBeanValueState.clear();
                        }

                        // 如果完成授信占用得数据先到 已经存在状态中 这里需要进行处理 将对应得数据写道侧输出流
                        DwdCreditOccupyBean creditOccupyBean = creditOccupyBeanValueState.value();
                        if (creditOccupyBean != null) {
                            creditOccupyBean.setApplyAmount(auditApprovalBean.getApplyAmount());
                            creditOccupyBean.setReplyAmount(auditApprovalBean.getReplyAmount());
                            ctx.output(occupyBeanTag, creditOccupyBean);
                            creditOccupyBeanValueState.clear();
                        }

                        // 如果完成合同制作得数据先到 已经存在状态中 这里需要进行处理 将对应得数据写道侧输出流
                        DwdLeaseContractProducedBean contractProducedBean = leaseContractProducedBeanValueState.value();
                        if (contractProducedBean != null) {
                            contractProducedBean.setApplyAmount(auditApprovalBean.getApplyAmount());
                            contractProducedBean.setReplyAmount(auditApprovalBean.getReplyAmount());
                            ctx.output(contractProducedBeanTag, contractProducedBean);
                            leaseContractProducedBeanValueState.clear();
                        }
                    } else {
                        replyBeanValueState.update(auditReplyBean);
                    }
                } else {
                    // credit 授信表
                    // 获取type 判断是否为新增
                    String type = jsonObject.getString("type");
                    DwdAuditApprovalBean auditApprovalBean = approvalBeanValueState.value();
                    if ("insert".equals(type)) {
                        // 此时为新增授信数据
                        DwdCreditAddBean creditAddBean = jsonObject.getObject("data", DwdCreditAddBean.class);
                        if (auditApprovalBean != null) {
                            // 审批通过得数据先到
                            if (auditApprovalBean.getReplyAmount() == null) {
                                // 批复得数据未到
                                creditAddBeanValueState.update(creditAddBean);
                            } else {
                                // 批复得数据也到了 审批得数据完整
                                creditAddBean.setApplyAmount(auditApprovalBean.getApplyAmount());
                                creditAddBean.setReplyAmount(auditApprovalBean.getReplyAmount());
                                creditAddBean.setAddTime(auditApprovalBean.getApproveTime());
                                ctx.output(creditAddBeanTag, creditAddBean);
                                approvalBeanValueState.clear();
                            }
                        }
                    } else {
                        String status = jsonObject.getJSONObject("data").getString("status");
                        if ("2".equals(status)) {
                            // 此时为授信占用数据
                            DwdCreditOccupyBean creditOccupyBean = jsonObject.getObject("data", DwdCreditOccupyBean.class);
                            creditOccupyBean.setOccupyTime(jsonObject.getJSONObject("data").getString("credit_occupy_time"));

                            if (auditApprovalBean != null) {
                                // 审批通过得数据先到
                                if (auditApprovalBean.getReplyAmount() == null) {
                                    // 批复得数据未到
                                    creditOccupyBeanValueState.update(creditOccupyBean);
                                } else {
                                    // 批复得数据也到了 审批得数据完整
                                    creditOccupyBean.setApplyAmount(auditApprovalBean.getApplyAmount());
                                    creditOccupyBean.setReplyAmount(auditApprovalBean.getReplyAmount());
                                    ctx.output(occupyBeanTag, creditOccupyBean);
                                    approvalBeanValueState.clear();
                                }
                            }

                        } else if ("3".equals(status)) {
                            // 此时为制作合同
                            DwdLeaseContractProducedBean contractProducedBean = jsonObject.getObject("data", DwdLeaseContractProducedBean.class);
                            contractProducedBean.setId(jsonObject.getJSONObject("data").getString("contract_id"));
                            contractProducedBean.setCreditId(jsonObject.getJSONObject("data").getString("id"));
                            contractProducedBean.setProducedTime(jsonObject.getJSONObject("data").getString("contract_produce_time"));

                            if (auditApprovalBean != null) {
                                // 审批通过得数据先到
                                if (auditApprovalBean.getReplyAmount() == null) {
                                    // 批复得数据未到
                                    leaseContractProducedBeanValueState.update(contractProducedBean);
                                } else {
                                    // 批复得数据也到了 审批得数据完整
                                    contractProducedBean.setApplyAmount(auditApprovalBean.getApplyAmount());
                                    contractProducedBean.setReplyAmount(auditApprovalBean.getReplyAmount());
                                    ctx.output(contractProducedBeanTag, contractProducedBean);
                                    approvalBeanValueState.clear();
                                }
                            }
                        }
                    }
                }
            }
        });

        // 获取侧输出流
        SideOutputDataStream<DwdAuditCancelBean> cancelStream = processStream.getSideOutput(cancelStreamTag);
        SideOutputDataStream<DwdAuditRejectBean> rejectStream = processStream.getSideOutput(rejectStreamTag);

        String approveTopic = "financial_dwd_audit_approve";
        String cancelTopic = "financial_dwd_audit_cancel";
        String rejectTopic = "financial_dwd_audit_reject";

        // 6.1 审批通过
        processStream.map(new MapFunction<DwdAuditApprovalBean, String>() {
                    @Override
                    public String map(DwdAuditApprovalBean value) throws Exception {
                        return JSON.toJSONString(value);
                    }
                })
                .sinkTo(KafkaUtil.getKafkaProducer(approveTopic,approveTopic+"_sink"))
                .name("approve_stream_sink");
        // 6.2 审批拒绝
        rejectStream.map(new MapFunction<DwdAuditRejectBean, String>() {
                    @Override
                    public String map(DwdAuditRejectBean value) throws Exception {
                        return JSON.toJSONString(value);
                    }
                })
                .sinkTo(KafkaUtil.getKafkaProducer(rejectTopic,rejectTopic+"_sink"))
                .name("reject_stream_sink");
        // 6.3 审批取消
        cancelStream.map(new MapFunction<DwdAuditCancelBean, String>() {
                    @Override
                    public String map(DwdAuditCancelBean value) throws Exception {
                        return JSON.toJSONString(value);
                    }
                })
                .sinkTo(KafkaUtil.getKafkaProducer(cancelTopic,cancelTopic+"_sink"))
                .name("cancel_stream_sink");

        // 将授信域写出到kafka对应得主题
        String creditAddTopic = "financial_dwd_audit_add";
        String creditOccupyTopic = "financial_dwd_audit_occupy";
        SideOutputDataStream<DwdCreditAddBean> creditAddStream = processStream.getSideOutput(creditAddBeanTag);
        SideOutputDataStream<DwdCreditOccupyBean> creditOccupyStream = processStream.getSideOutput(occupyBeanTag);

        // 7.1 新增授信数据
        creditAddStream.map(new MapFunction<DwdCreditAddBean, String>() {
                    @Override
                    public String map(DwdCreditAddBean value) throws Exception {
                        return JSON.toJSONString(value);
                    }
                })
                .sinkTo(KafkaUtil.getKafkaProducer(creditAddTopic,creditAddTopic+"_sink"))
                .name("credit_add_sink");

        // 7.2 完成授信占用
        creditOccupyStream.map(new MapFunction<DwdCreditOccupyBean, String>() {
                    @Override
                    public String map(DwdCreditOccupyBean value) throws Exception {
                        return JSON.toJSONString(value);
                    }
                })
                .sinkTo(KafkaUtil.getKafkaProducer(creditOccupyTopic,creditOccupyTopic+"_sink"))
                .name("credit_occupy_sink");

        // 8过滤合同制作状态得授信数据
        String leaseContractProduce = "financial_dwd_lease_contract_produce";
        SideOutputDataStream<DwdLeaseContractProducedBean> contractProdureStream = processStream.getSideOutput(contractProducedBeanTag);
        contractProdureStream.map(new MapFunction<DwdLeaseContractProducedBean, String>() {
                    @Override
                    public String map(DwdLeaseContractProducedBean value) throws Exception {
                        return JSON.toJSONString(value);
                    }
                })
                .sinkTo(KafkaUtil.getKafkaProducer(leaseContractProduce,leaseContractProduce+"_sink"))
                .name("contract_produce_sink");

        // 9 合并租赁域合同制作数据和合同表数据
        KeyedStream<DwdLeaseContractProducedBean, String> producedBeanStringKeyedStream = contractProdureStream.keyBy(new KeySelector<DwdLeaseContractProducedBean, String>() {
            @Override
            public String getKey(DwdLeaseContractProducedBean value) throws Exception {
                return value.getCreditId();
            }
        });

        KeyedStream<JSONObject, String> contractKeyedStream = contractStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getJSONObject("data").getString("credit_id");
            }
        });

        ConnectedStreams<DwdLeaseContractProducedBean, JSONObject> connectedStreams = producedBeanStringKeyedStream.connect(contractKeyedStream);

        // 10 数据分流
        OutputTag<DwdLeaseExecutionBean> executionTag= new OutputTag<DwdLeaseExecutionBean>("execution_tag"){};

        // 10.1 获取主流 签约流
        SingleOutputStreamOperator<DwdLeaseSignedBean> signedBeanStream = connectedStreams.process(new KeyedCoProcessFunction<String, DwdLeaseContractProducedBean, JSONObject, DwdLeaseSignedBean>() {

            //合同制作
            private ValueState<DwdLeaseContractProducedBean> contractProducedState;

            //签约数据
            private ValueState<DwdLeaseSignedBean> signState;

            // 起租数据
            private ValueState<DwdLeaseExecutionBean> executionState;

            @Override
            public void open(Configuration parameters) throws Exception {
                contractProducedState = getRuntimeContext().getState(new ValueStateDescriptor<DwdLeaseContractProducedBean>("contract_produced_state", DwdLeaseContractProducedBean.class));
                signState = getRuntimeContext().getState(new ValueStateDescriptor<DwdLeaseSignedBean>("sign_state", DwdLeaseSignedBean.class));
                executionState = getRuntimeContext().getState(new ValueStateDescriptor<DwdLeaseExecutionBean>("execution_state", DwdLeaseExecutionBean.class));
            }

            @Override
            public void processElement1(DwdLeaseContractProducedBean bean, KeyedCoProcessFunction<String, DwdLeaseContractProducedBean, JSONObject, DwdLeaseSignedBean>.Context ctx, Collector<DwdLeaseSignedBean> out) throws Exception {
                // 当前为合同制作数据 判断合同表数据是否先到 进行处理
                DwdLeaseSignedBean signedBean = signState.value();
                DwdLeaseExecutionBean executionBean = executionState.value();

                if (signedBean == null || executionBean == null) {
                    contractProducedState.update(bean);
                }

                // 如果签约数据不为空
                if (signedBean != null) {
                    //此时签约得数据完整 需要写出到主题
                    signedBean.setCreditFacilityId(bean.getCreditFacilityId());
                    signedBean.setApplyAmount(bean.getReplyAmount());
                    signedBean.setReplyAmount(bean.getReplyAmount());
                    signedBean.setCreditAmount(bean.getCreditAmount());
                    out.collect(signedBean);
                    signState.clear();
                }

                // 如果起租数据不为空
                if (executionBean != null) {
                    //此时起租数据完整 需要写出到侧输出流
                    executionBean.setCreditFacilityId(bean.getCreditFacilityId());
                    executionBean.setApplyAmount(bean.getApplyAmount());
                    executionBean.setReplyAmount(bean.getReplyAmount());
                    executionBean.setCreditAmount(bean.getCreditAmount());
                    ctx.output(executionTag, executionBean);
                    executionState.clear();
                }
            }

            @Override
            public void processElement2(JSONObject jsonObject, KeyedCoProcessFunction<String, DwdLeaseContractProducedBean, JSONObject, DwdLeaseSignedBean>.Context ctx, Collector<DwdLeaseSignedBean> out) throws Exception {
                // 当前为合同表代表签约和起租 判断合同制作数据是否先到进行处理
                DwdLeaseContractProducedBean contractProducedBean = contractProducedState.value();

                String type = jsonObject.getString("type");
                if ("update".equals(type)) {
                    String status = jsonObject.getJSONObject("data").getString("status");
                    if ("2".equals(status)) {
                        // 当前为签约得数据
                        DwdLeaseSignedBean leaseSignedBean = jsonObject.getObject("data", DwdLeaseSignedBean.class);
                        if (contractProducedBean == null) {
                            signState.update(leaseSignedBean);
                        } else {
                            // 合同制作数据先到 需要将签约数据写出到主流
                            leaseSignedBean.setCreditFacilityId(contractProducedBean.getCreditFacilityId());
                            leaseSignedBean.setApplyAmount(contractProducedBean.getReplyAmount());
                            leaseSignedBean.setReplyAmount(contractProducedBean.getReplyAmount());
                            leaseSignedBean.setCreditAmount(contractProducedBean.getCreditAmount());
                            out.collect(leaseSignedBean);
                        }
                    } else if ("3".equals(status)) {
                        // 当前为起租数据
                        DwdLeaseExecutionBean executionBean = jsonObject.getObject("data", DwdLeaseExecutionBean.class);
                        if (contractProducedBean == null) {
                            executionState.update(executionBean);
                        } else {
                            // 合同制作数据先到 需要将起租数据写出到测输出流
                            executionBean.setCreditFacilityId(contractProducedBean.getCreditFacilityId());
                            executionBean.setApplyAmount(contractProducedBean.getApplyAmount());
                            executionBean.setReplyAmount(contractProducedBean.getReplyAmount());
                            executionBean.setCreditAmount(contractProducedBean.getCreditAmount());
                            ctx.output(executionTag, executionBean);
                        }
                    }
                }
            }
        });

        // 10.2 获取起租流
        SideOutputDataStream<DwdLeaseExecutionBean> executionStream = signedBeanStream.getSideOutput(executionTag);

        // 11 提取起租流 将签约和起租数据写出到对应得Kafka主题
        String signedTopic = "financial_dwd_lease_sign";
        String executionTopic = "financial_dwd_lease_execution";
        // 11. 写出签约数据
        signedBeanStream.map(new MapFunction<DwdLeaseSignedBean, String>() {
                    @Override
                    public String map(DwdLeaseSignedBean value) throws Exception {
                        return JSON.toJSONString(value);
                    }
                })
                .sinkTo(KafkaUtil.getKafkaProducer(signedTopic,signedTopic+"_sink"))
                .name("sign_sink");

        // 11.2 写出起租数据
        executionStream.map(new MapFunction<DwdLeaseExecutionBean, String>() {
                    @Override
                    public String map(DwdLeaseExecutionBean value) throws Exception {
                        return JSON.toJSONString(value);
                    }
                })
                .sinkTo(KafkaUtil.getKafkaProducer(executionTopic,executionTopic+"_sink"))
                .name("execution_sink");

        //12 运行程序
        env.execute();
    }
}
